package datastream.api.operator.group.source;

import datastream.api.operator.group.pojo.ReduceModel;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class ReduceSource implements SourceFunction<ReduceModel> {
    @Override
    public void run(SourceContext<ReduceModel> ctx) throws Exception {
        ctx.collect(ReduceModel.builder().produceId(1).income(1.0).build());
        Thread.sleep(1000);

        ctx.collect(ReduceModel.builder().produceId(2).income(2.0).build());
        Thread.sleep(1000);

        ctx.collect(ReduceModel.builder().produceId(3).income(3.0).build());
        Thread.sleep(1000);

        ctx.collect(ReduceModel.builder().produceId(2).income(20.0).build());
        Thread.sleep(1000);

        ctx.collect(ReduceModel.builder().produceId(1).income(10.0).build());
        Thread.sleep(1000);

        ctx.collect(ReduceModel.builder().produceId(3).income(30.0).build());
    }

    @Override
    public void cancel() {

    }
}
